package com.jltx.zeus.observable

import io.reactivex.ObservableOnSubscribe
import io.reactivex.disposables.Disposable
import io.reactivex.functions.Cancellable

/**
 * @author jltxseo
 *         Created by junlintianxia on 2020/05/20.
 */
abstract class ObservableOnSubscribeEx<T> :
    ObservableOnSubscribe<T> {

    @Throws(Exception::class)
    override fun subscribe(emitter: io.reactivex.ObservableEmitter<T>) {
        val createEmitter: ObservableEmitter<T> = ObservableEmitter(emitter)
        subscribe(createEmitter)
    }

    @Throws(Exception::class)
    abstract fun subscribe(emitter: ObservableEmitter<T>)

    inner class ObservableEmitter<T>(private val mEmitter: io.reactivex.ObservableEmitter<T>) :
        io.reactivex.ObservableEmitter<T> {
        @Volatile
        var mTerminated = false
        override fun setDisposable(d: Disposable?) {
            mEmitter.setDisposable(d)
        }

        override fun setCancellable(c: Cancellable?) {
            mEmitter.setCancellable(c)
        }

        override fun isDisposed(): Boolean {
            return mEmitter.isDisposed
        }

        override fun serialize(): io.reactivex.ObservableEmitter<T> {
            return mEmitter.serialize()
        }

        override fun tryOnError(t: Throwable): Boolean {
            return mEmitter.tryOnError(t)
        }

        override fun onNext(value: T) {
            if (isDisposed || mTerminated) return
            mEmitter.onNext(value)
        }

        override fun onError(error: Throwable) {
            if (isDisposed || mTerminated) return
            mTerminated = true
            mEmitter.onError(error)
        }

        override fun onComplete() {
            if (isDisposed || mTerminated) return
            mTerminated = true
            mEmitter.onComplete()
        }

    }
}
